package com.liuyu.common.framework.rocketmq.service.spring;

import com.alibaba.csp.sentinel.annotation.SentinelResource;
import com.alibaba.fastjson.JSON;
import com.liuyu.common.framework.common.constant.AppConstants;
import com.liuyu.common.framework.common.distribute.RedissonLockUtils;
import com.liuyu.common.framework.common.utils.GlobalExecutorUtils;
import com.liuyu.common.framework.lang.exception.BusinessException;
import com.liuyu.common.framework.rocketmq.fallback.RocketFallbackMessage;
import com.liuyu.common.framework.rocketmq.service.facade.MessageProducerService;
import com.liuyu.common.framework.rocketmq.service.sentinel.RocketSentinelCommHandler;
import com.liuyu.common.framework.rocketmq.vo.MqMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @ClassName RocketMqMessageProducerServiceImpl
 * @Author: liuyu
 * @Date: Created in 2021/7/29 14:59
 * @Version: 1
 * @Modified By:
 */
@Service
@Slf4j
public class RocketMqMessageProducerServiceImpl implements MessageProducerService, Runnable, SmartInitializingSingleton {
    private static final String YYYY_MM_DD_HH_MM_SS = "yyyy-MM-dd HH:mm:ss";
    private static final String ROCKET_FAIL_MESSAGE = "消息发送失败: 主题:{},标签：{},消息:{} [错误:{},{}]";
    private static final String SEND_MESSAGE = "sendMessage";
    private static final String SEND_ORDER_MESSAGE = "sendOrderMessage";
    private String charset = "UTF-8";
    @Autowired
    private RocketMQTemplate rocketMqTemplate;

    @Autowired
    private RedisTemplate redisTemplate;

    public static final String ROCKET_FALL_BACK_LIST = AppConstants.APP_KEY + ":rocket:fall:back:list";
    public static final String ROCKET_FALL_BACK_LIST_LOCK = AppConstants.APP_KEY + ":lock:rocket:fall:back:list:lock";

    public static AtomicBoolean health = new AtomicBoolean(true);

    Map<String, Method> methodMap = new ConcurrentHashMap<>();

    @Autowired
    private RedissonClient redissonClient;


    /**
     * 发送普通的消息
     *
     * @param topic
     * @param message
     * @return void
     * @author liuyu
     * @date 2021/7/29 15:00
     */
    @Override
    /**
     * 热点参数必须使用注解才能生效。
     *
     * @return
     * @author liuyu
     * @date 2022/3/21 15:01
     */
    @SentinelResource(value = "sendMessage",
            blockHandlerClass = RocketSentinelCommHandler.class,
            blockHandler = "sendMessageFallback",
            fallbackClass = RocketSentinelCommHandler.class,
            fallback = "sendMessageFallback"
    )
    public <T> void sendMessage(String topic, MqMessage<T> message) throws Exception {
        long start = System.currentTimeMillis();
        String json = JSON.toJSONStringWithDateFormat(message, YYYY_MM_DD_HH_MM_SS);
        String tags = null;
        if (StringUtils.isNotBlank(message.getTags())) {
            tags = message.getTags();
        }
        Message message1 = new Message(topic, tags, json.getBytes(StandardCharsets.UTF_8));
        String levelName = null;
        if (message.getLevel() != null) {
            message1.setDelayTimeLevel(message.getLevel().getLevel());
            levelName = message.getLevel().getName();
        }
//        throw new BusinessException("模拟投递失败");
        SendResult sendResult = rocketMqTemplate.getProducer().send(message1);
        if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            health.set(true);
            if (StringUtils.isNotBlank(levelName)) {
                log.info("延迟队列发送成功:主题{},标签：{}, 消息体： {}, messageId = {},延迟等级:{},耗时:{}", topic, tags, message, sendResult.getMsgId(), levelName, (System.currentTimeMillis() - start));
            } else {
                log.info("普通的消息发送成功：主题：{}, 标签：{}, messageId:{}, 耗时:{}", topic, tags, sendResult.getMsgId(), (System.currentTimeMillis() - start));
            }
        } else {
            health.set(false);
            if (StringUtils.isNotBlank(levelName)) {
                log.warn("延迟队列发送失败:主题{},标签：{}, 消息体： {}, messageId = {},延迟等级:{},耗时:{}", topic, tags, message, sendResult.getMsgId(), levelName, (System.currentTimeMillis() - start));
            } else {
                log.warn("普通的消息发送失败：主题：{}, 标签：{}, messageId:{}, 耗时:{}", topic, tags, sendResult.getMsgId(), (System.currentTimeMillis() - start));
            }
        }


    }

    /**
     * 普通消息熔断
     *
     * @param topic
     * @param message
     * @param e
     * @return void
     * @author liuyu
     * @date 2022/4/11 21:53
     */
    private <T> void sendMessageFallback(String topic, MqMessage<T> message, Throwable e) {
        log.error(ROCKET_FAIL_MESSAGE, topic, message.getTags(), message, e.getMessage(), e);
        //添加到redis中去
        redisTemplate.opsForList().leftPush(ROCKET_FALL_BACK_LIST, new RocketFallbackMessage(SEND_MESSAGE, topic, null, null, message));
        health.set(false);
    }


    /**
     * 发送局部顺序的消息
     *
     * @param topic
     * @param message
     * @return void
     * @throws
     * @author liuyu
     * @date 2022/11/30 15:39
     * //    @SentinelResource(fallback = "sendOrderMessageFallBack")
     */
    @Override
    /**
     * 热点参数必须使用注解才能生效。
     *
     * @return
     * @author liuyu
     * @date 2022/3/21 15:01
     */
    @SentinelResource(value = "sendOrderMessage",
            blockHandlerClass = RocketSentinelCommHandler.class,
            blockHandler = "sendOrderMessageFallBack",
            fallbackClass = RocketSentinelCommHandler.class,
            fallback = "sendOrderMessageFallBack"
    )
    public <T> void sendOrderMessage(String topic, MqMessage<T> message) throws Exception {
        String json = JSON.toJSONStringWithDateFormat(message, YYYY_MM_DD_HH_MM_SS);
        long start = System.currentTimeMillis();
        String tags = null;
        if (StringUtils.isNotBlank(message.getTags())) {
            tags = message.getTags();
        }

        String key = null;
        if (StringUtils.isNotBlank(message.getKey())) {
            key = message.getKey();
        } else {
            throw new BusinessException("顺序消息：key不能为空");
        }
        Message message1 = new Message(topic, tags, key, json.getBytes(StandardCharsets.UTF_8));
        String levelName = null;
        if (message.getLevel() != null) {
            message1.setDelayTimeLevel(message.getLevel().getLevel());
            levelName = message.getLevel().getName();
        }
        SendResult sendResult = rocketMqTemplate.getProducer().send(message1, new SelectMessageQueueByHash(), key);
        health.set(true);
        if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            health.set(true);
            if (StringUtils.isNotBlank(levelName)) {
                log.info("顺序延迟消息发送成功:主题{},标签：{}, 消息体： {}, messageId = {},延迟等级:{},key:{},队列：{},耗时:{}", topic, tags, message, sendResult.getMsgId(), levelName, message.getKey(), sendResult.getMessageQueue().getQueueId(), (System.currentTimeMillis() - start));
            } else {
                log.info("顺序消息发送成功：主题：{}, 标签：{}, messageId:{},key:{},队列：{}, 耗时:{}", topic, tags, sendResult.getMsgId(), message.getKey(), sendResult.getMessageQueue().getQueueId(), (System.currentTimeMillis() - start));
            }
        } else {
            health.set(false);
            if (StringUtils.isNotBlank(levelName)) {
                log.warn("顺序延迟消息发送失败:主题{},标签：{}, 消息体： {}, messageId = {},延迟等级:{},key:{},耗时:{}", topic, tags, message, sendResult.getMsgId(), levelName, message.getKey(), (System.currentTimeMillis() - start));
            } else {
                log.warn("顺序消息发送失败：主题：{}, 标签：{}, messageId:{},key:{}, 耗时:{}", topic, tags, sendResult.getMsgId(), message.getKey(), (System.currentTimeMillis() - start));
            }
        }

    }

    /**
     * 顺序消息熔断
     *
     * @param topic
     * @param message
     * @param e
     * @return void
     * @author liuyu
     * @date 2022/4/11 21:52
     */
    private <T> void sendOrderMessageFallBack(String topic, MqMessage<T> message, Throwable e) {
        log.error(ROCKET_FAIL_MESSAGE, topic, message.getTags(), message, e.getMessage(), e);
        //添加到redis中去
        redisTemplate.opsForList().leftPush(ROCKET_FALL_BACK_LIST, new RocketFallbackMessage(SEND_ORDER_MESSAGE, topic, null, null, message));
        health.set(false);
    }


    /**
     * 发送事务消息
     *
     * @param topicName
     * @param message
     * @param object
     * @param header
     * @return org.apache.rocketmq.client.producer.TransactionSendResult
     * @author liuyu
     * @date 2021/12/23 11:05
     */
    @Override
    /**
     * 热点参数必须使用注解才能生效。
     *
     * @return
     * @author liuyu
     * @date 2022/3/21 15:01
     */
    @SentinelResource(value = "sendMessageInTransaction",
            blockHandlerClass = RocketSentinelCommHandler.class,
            blockHandler = "sendMessageInTransactionFallBack",
            fallbackClass = RocketSentinelCommHandler.class,
            fallback = "sendMessageInTransactionFallback"
    )
    public <T> TransactionSendResult sendMessageInTransaction(String topicName, MqMessage<T> message, Object object, Map<String, Object> header) {
        String messages = JSON.toJSONString(message);
        String topic;
        if (StringUtils.isNotBlank(message.getTags())) {
            topic = String.format("%s:%s", topicName, message.getTags());
        } else {
            topic = topicName;
        }
//        MessageConst.PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS, "0"  设置检测时间间隔
        org.springframework.messaging.Message message1;
        MessageBuilder messageBuilder = MessageBuilder.withPayload(messages);
        for (Map.Entry<String, Object> entry : header.entrySet()) {
            messageBuilder.setHeader(entry.getKey(), entry.getValue());
        }
        message1 = messageBuilder.build();
//                .setHeader(MessageConst.PROPERTY_CHECK_IMMUNITY_TIME_IN_SECONDS, 5)

        TransactionSendResult transactionSendResult = rocketMqTemplate.sendMessageInTransaction(topic, message1, object);
        health.set(true);
        log.info("事务消息-本地事务执行状态：{},{}" + transactionSendResult.toString(), transactionSendResult.getLocalTransactionState().name());
        return transactionSendResult;
    }


    /**
     * 事务消息熔断
     *
     * @param topic
     * @param message
     * @param e
     * @return org.apache.rocketmq.client.producer.TransactionSendResult
     * @author liuyu
     * @date 2022/4/11 21:52
     */
    private <T> TransactionSendResult sendMessageInTransactionFallback(String topic, MqMessage<T> message, Object object, Map<String, Object> header, Throwable e) {
        log.error(ROCKET_FAIL_MESSAGE, topic, message.getTags(), message, e.getMessage(), e);
        //添加到redis中去
        redisTemplate.opsForList().leftPush(ROCKET_FALL_BACK_LIST, new RocketFallbackMessage("sendDelayMessageFallback", topic, null, null, message));
        TransactionSendResult sendResult = new TransactionSendResult();
        sendResult.setLocalTransactionState(LocalTransactionState.UNKNOW);
        health.set(false);
        return sendResult;
    }

    /**
     * When an object implementing interface <code>Runnable</code> is used
     * to create a thread, starting the thread causes the object's
     * <code>run</code> method to be called in that separately executing
     * thread.
     * <p>
     * The general contract of the method <code>run</code> is that it may
     * take any action whatsoever.
     *
     * @see Thread#run()
     */
    @Override
    public void run() {
        if (health.get()) {
            //增加分布式锁来防止重复获取并投递
            RLock lock = null;
            try {
                lock = RedissonLockUtils.lock(ROCKET_FALL_BACK_LIST_LOCK, TimeUnit.MINUTES, 5);
                Long size = redisTemplate.opsForList().size(ROCKET_FALL_BACK_LIST);
                for (int i = 0; i < size; i++) {
                    //从邮编获取出来
                    RocketFallbackMessage<MqMessage> rocketFallbackMessage = (RocketFallbackMessage<MqMessage>) redisTemplate.opsForList().rightPop(ROCKET_FALL_BACK_LIST);
                    log.info("消息服务器恢复，从redis获取重发", rocketFallbackMessage);
                    Method method = methodMap.get(rocketFallbackMessage.getMethod());
                    if (method != null) {
                        try {
                            method.invoke(this, rocketFallbackMessage.getMessage().getTopic(), rocketFallbackMessage.getMessage());
                        } catch (Exception e) {
                            log.error("重新发送rocket失败：{},错误：{}", rocketFallbackMessage, e.getMessage(), e);
                            redisTemplate.opsForList().rightPush(ROCKET_FALL_BACK_LIST, rocketFallbackMessage);
                            break;
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                //lock
                RedissonLockUtils.unlock(lock);
            }

        }

    }

    /**
     * Invoked right at the end of the singleton pre-instantiation phase,
     * with a guarantee that all regular singleton beans have been created
     * already. {@link ListableBeanFactory#getBeansOfType} calls within
     * this method won't trigger accidental side effects during bootstrap.
     * <p><b>NOTE:</b> This callback won't be triggered for singleton beans
     * lazily initialized on demand after {@link BeanFactory} bootstrap,
     * and not for any other bean scope either. Carefully use it for beans
     * with the intended bootstrap semantics only.
     */
    @Override
    public void afterSingletonsInstantiated() {
        Method[] method = getClass().getDeclaredMethods();
        for (Method method1 : method) {
            methodMap.put(method1.getName(), method1);
        }
        //5分钟执行一次重投
        GlobalExecutorUtils.scheduleAtFixedRate(this, 0, 5, TimeUnit.MINUTES);
        RedissonLockUtils.setRedissonClient(redissonClient);
    }
}
